package com.bleeth.flow.step.output;


import com.bleeth.flow.core.common.APlugin;
import com.bleeth.flow.core.common.PluginAnnotation;
import com.bleeth.flow.core.common.PluginTypeEnum;
import com.jd.platform.async.wrapper.WorkerWrapper;
import lombok.Data;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

import java.io.Serializable;
import java.util.Map;
import java.util.Properties;

/**
 * @author ：Bleeth
 * @date ：2021-08-19
 * @description：
 */
@Data
@PluginAnnotation(name = "mysql写",
        type = PluginTypeEnum.OUTPUT,
        description = "",
        id = "MySQLWriterPlugin")
public class MySQLWriterPlugin extends APlugin implements Serializable {

    private static final long serialVersionUID = -33489724224931846L;


    private String url;
    private String table;
    private String user;
    private String password;


    @Override
    public Dataset<Row> action(Dataset<Row> param, Map<String, WorkerWrapper> allWrappers) {
        super.action(param, allWrappers);

        String fromPluginName = fromList.get(0);
        WorkerWrapper fromWrapper = allWrappers.get(fromPluginName);
        Dataset<Row> ds = (Dataset<Row>) fromWrapper.getWorkResult().getResult();

        Properties connectionProperties = new Properties();
        connectionProperties.put("user", user);
        connectionProperties.put("password", password);
        connectionProperties.put("driver", "com.mysql.cj.jdbc.Driver");

        ds.write().mode("append").jdbc(url, table, connectionProperties);

        return null;
    }

}
